package com.exd.flink.source;

import java.util.Date;
import java.util.Properties;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import com.exd.flink.common.PropertiesTool;

/**
 * 描述: kafka数据源
 * 创建人: 赵兴炎
 * 日期: 2022年9月13日
 */
public class KafkaSource {
	
	public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(){
		String bootstrapServer = PropertiesTool.getParameterTool().get("exd.kafka.bootstrap.servers");
		String tocpicName=PropertiesTool.getParameterTool().get("exd.kafka.tocpic.name");
		String groupId=PropertiesTool.getParameterTool().get("exd.kafka.group.id");
		if(!PropertiesTool.getParameterTool().getBoolean("exd.kafka.tocpic.name")) {
			groupId=String.valueOf(new Date().getTime());
		}
		System.out.println("ip:"+bootstrapServer+",tocpicName:"+tocpicName+",groupId:"+groupId);
		Properties properties = new Properties();
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,  bootstrapServer);
		//一次拉取的最小可返回数据量 byte字节单位 设置1kb
//		properties.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024);
		//一次拉取的最大数据量：byte字节单位 设置 200m
//		properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 200 * 1024 * 1024);
//		//每次Poll的最大数量
//		properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10000);
//		//一次拉取的最大等待时间 毫秒
//		properties.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 1000*60*10);
		//一次拉取时，每个分区最大拉取数据量，默认1M
//		properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 40 * 1024 * 1024);
//		properties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
		properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
		FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<String>(tocpicName, new SimpleStringSchema(), properties);
		flinkKafkaConsumer.setStartFromGroupOffsets();
		return flinkKafkaConsumer;
	}

}
